package org.example.config;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.example.util.MapStringToMapUtil;
import org.jetbrains.annotations.NotNull;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * 消息手动接收确认机制
 *
 *@ClassName: MyAckReceiver
 *@Author: CJ
 *@Date: 2021-9-24 14:33
 */
@Slf4j
@Component
public class MyAckReceiver implements ChannelAwareMessageListener {

    @Override
    public void onMessage(@NotNull Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            // 因为传递消息的时候用的map传递，所以将Map从Message内取出需要做些处理
            String msg = message.toString();
            String[] msgArray = msg.split("'");
            Map<String, String> msgMap = MapStringToMapUtil.getStringMap(msgArray[1].trim());
            String messageId = msgMap.get("messageId");
            String messageData = msgMap.get("messageData");
            String createTime = msgMap.get("createTime");
            log.info("MyAckReceiver  messageId:" + messageId + "  messageData:" + messageData + "  createTime:" + createTime);
            log.info("消费的主题消息来自：" + message.getMessageProperties().getConsumerQueue());
            // 第二个参数，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息
            channel.basicAck(deliveryTag, true);
            // 第二个参数，true会重新放回队列
            /*channel.basicReject(deliveryTag, true);*/
        } catch (Exception e) {
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }
    }
}
